package com.wyz.socket.core.impl.dispatcher;

import com.wyz.socket.core.api.dispatcher.SendDispatcher;
import com.wyz.socket.core.base.IoArgs;
import com.wyz.socket.core.api.Sender;
import com.wyz.socket.core.base.exceptions.EmptyIoArgsException;
import com.wyz.socket.core.base.packet.AbstractSendPacket;

import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncSendDispatcher implements SendDispatcher, IoArgs.IoBufferEventProcessor, AsyncPacketReader.PacketProvider {

    private final BlockingQueue<AbstractSendPacket<?>> queue = new ArrayBlockingQueue<AbstractSendPacket<?>>(16);
    private final AtomicBoolean isSending = new AtomicBoolean(false);
    private final AsyncPacketReader reader;
    private final Sender sender;
    private volatile long activeTime = System.currentTimeMillis();
    private final AtomicBoolean isClosed = new AtomicBoolean();

    public AsyncSendDispatcher(Sender sender) {
        this.sender = sender;
        this.sender.registerSocketListener(this);
        this.sender.setSendListener(this);
        this.reader = new AsyncPacketReader(this);

    }

    private void requestSendPacket(boolean isCallFromConsume) {
        synchronized (isSending) {
            final AtomicBoolean isSending = this.isSending;
            final boolean oldState = isSending.get();
            if ((oldState && !isCallFromConsume) || isClosed.get()) {
                return;
            }
            if (isCallFromConsume && !oldState) {
                throw new IllegalStateException("");
            }
            if (reader.requestTakePacket()) {
                isSending.set(true);
                try {
                    sender.postSendAsync();
                } catch (Exception e) {
                    e.printStackTrace();
                    close();
                }
            } else {
                isSending.set(false);
            }
        }
    }

    @Override
    public long getLastActiveTime() {
        return activeTime;
    }

    @Override
    public void onstart() {
        activeTime = System.currentTimeMillis();
    }

    @Override
    public IoArgs providerIoArgs() {
        if (isClosed.get()) return null;
        return reader.createIoArgsAndFillData();
    }

    @Override
    public boolean onConsumeFailed(Throwable e) {
        e.printStackTrace();
        if (e instanceof EmptyIoArgsException) {
            requestSendPacket(true);
            return false;
        } else {
            return true;
        }
    }

    @Override
    public boolean onConsumeCompleted(IoArgs args) {
        synchronized (isSending) {
            AtomicBoolean isSending = this.isSending;
            final boolean isRunning = !isClosed.get();
            if (!isSending.get() && isRunning) {
                throw new IllegalStateException("");
            }
            isSending.set(isRunning && reader.requestTakePacket());
            return isSending.get();
        }
    }

    @Override
    public void onEnd() {

    }

    @Override
    public void send(AbstractSendPacket<?> packet) {
        try {
            this.queue.put(packet);
            requestSendPacket(false);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    @Override
    public void sendHeartbeat() {
        if (!this.queue.isEmpty()) {
            return;
        }
        if (reader.requestSendHeartbeatFrame()) {
            requestSendPacket(false);
        }
    }

    @Override
    public void cancel(AbstractSendPacket<?> packet) {
        boolean remove = this.queue.remove(packet);
        if (remove) {
            packet.cancel();
            return;
        }
        reader.cancel(packet);
    }


    @Override
    public AbstractSendPacket<?> nextPacket() {
        AbstractSendPacket<?> packet = this.queue.poll();
        if (packet == null) {
            return null;
        }
        if (packet.isCancel()) {
            return nextPacket();
        }
        return packet;
    }

    @Override
    public void completeSendPacket(AbstractSendPacket<?> packet, boolean isSucceed) {
        try {
            packet.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Override
    public void close()  {
        try {
            sender.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public <T> void onClose(T t) {
        if (t instanceof SocketChannel) {
            if (!((SocketChannel) t).isOpen()) {
                if (isClosed.compareAndSet(false, true)) {
                    reader.close();
                    queue.clear();
                    synchronized (isSending) {
                        isSending.set(false);
                    }
                }
            }
        }
    }
}
